package com.sujie;


/**
 * @author: xinmi
 * @date: 2020/12/11
 * @description:
 */
import java.io.*;
import java.time.LocalDate;
import java.util.*;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumerDemo {

    private static String fileName;
    private static FileOutputStream fos;
    private static DataOutputStream outStream;

    private static boolean firstTime = true;

    public static void main(String[] args) throws FileNotFoundException {

        generateFileNameSchedule();

        if (fileName == null || fileName.isEmpty()) {
            LocalDate localDate = LocalDate.now();
            fileName = System.getProperty("user.dir") + "/data/gps/GPS" + localDate.toString().replace("-", "_") + ".txt";
            System.out.println("fileName =" + fileName);
        }

        // Set up client Java properties
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "127.0.0.1:9092");
        // Just a user-defined string to identify the consumer group
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        // Enable auto offset commit
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());

        fos = new FileOutputStream(fileName);
        outStream = new DataOutputStream(new BufferedOutputStream(fos));

        firstTime = false;
        
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            // List of topics to subscribe to
            consumer.subscribe(Arrays.asList("jt809"));
            while (true) {
                try {
                    ConsumerRecords<String, String> records = consumer.poll(100);
                    for (ConsumerRecord<String, String> record : records) {
/*                        System.out.printf("Offset = %d\n", record.offset());
                        System.out.printf("Key    = %s\n", record.key());
                        System.out.printf("Value  = %s\n", record.value());*/
                        //System.out.printf("%s\n", record.value());
                        outStream.writeUTF(record.value() + "\n");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private static void generateFileNameSchedule() {
        File dataDir = new File(System.getProperty("user.dir") + "/data");
        if (!dataDir.exists() && !dataDir.isDirectory())
            if (!dataDir.mkdir()) {
                System.out.println("mkdir data failed");
            }

        File gpsDir = new File(System.getProperty("user.dir") + "/data/gps");
        if (!gpsDir.exists() && !gpsDir.isDirectory())
            if (!gpsDir.mkdir()) {
                System.out.println("mkdir data/gps failed");
            }

        Timer timer = new Timer();
        Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.HOUR_OF_DAY, Config.GPS_HOUR); //4点
        calendar.set(Calendar.MINUTE, Config.GPS_MINUTE);//45分
        calendar.set(Calendar.SECOND, Config.GPS_SECOND);//0秒
        //每天的凌晨4:45,重新生成新文件
        timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                //每天定时生成新的文件路径和相应的输出流
                if(firstTime == false) {
                    LocalDate localDate = LocalDate.now();
                    fileName = System.getProperty("user.dir") + "/data/gps/GPS" + localDate.toString().replace("-", "_") + ".txt";
                    System.out.println("fileName =" + fileName);
                    try {
                        if (outStream != null){
                            outStream.close();
                        }
                        fos = new FileOutputStream(fileName);
                        outStream = new DataOutputStream(new BufferedOutputStream(fos));
                    } catch (Exception e) {
                        System.out.println("写文件内容操作出错");
                        e.printStackTrace();
                    }
                }
            }
        }, calendar.getTime(), 24 * 60 * 60 * 1000);
    }
}
